package xuecheng.tool.mq.alibaba.rocket;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

import javax.annotation.Nullable;

import org.apache.commons.lang3.StringUtils;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.core.annotation.AnnotationUtils;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import xuecheng.tool.mq.MqListener;

/**
 * 自定义 RocketMq 配置
 *
 * @author PengXueCheng
 * @version 1.0.0
 **/
@SpringBootConfiguration
@EnableConfigurationProperties(RocketMqProperties.class)
@ConditionalOnProperty(prefix = "cpx.rocketmq", value = "enable", havingValue = "true")
@Slf4j
public class CpxRocketMqBeanPostProcessor implements BeanPostProcessor {

    @Autowired
    private RocketMqProperties rocketMqProperties;

    /**
     * 注册消费者
     *
     * @param bean
     * @param beanName
     * @return
     * @throws BeansException
     */
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        // 对所有 MqListener 的所有子类且添加有 RocketMQMessageListener 类注解的做 RedisMq 主题订阅
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        RocketMqMessageListener listener = AnnotationUtils.findAnnotation(targetClass, RocketMqMessageListener.class);
        boolean assignableFrom = MqListener.class.isAssignableFrom(targetClass);
        boolean ok = listener != null && assignableFrom;
        if (ok) {
            register(listener, (MqListener)bean);
        }
        return bean;
    }

    /**
     * RocketMq 主题订阅
     *
     * @param listenerAnnotation
     *            队列类注解
     * @param mqListener
     *            队列消息监听接口
     */
    private void register(RocketMqMessageListener listenerAnnotation, MqListener mqListener) {
        String topic = listenerAnnotation.topic();
        String tag = listenerAnnotation.tag();
        String gid = listenerAnnotation.consumerGroup();
        String nameServer = listenerAnnotation.nameServer();
        if (StringUtils.isEmpty(nameServer)) {
            nameServer = rocketMqProperties.getNameServer();
        }

        Class messageClass = getMessageClass(mqListener);
        subscribe(gid, nameServer, topic, tag, messageClass, mqListener);
    }

    /**
     * rocketmq 主题订阅
     *
     * @param gid
     *            您在控制台创建的 Group ID
     * @param nameServerAddress
     *            设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看
     * @param topic
     *            消息一级分类
     * @param tag
     *            消息二级分类(可选, 默认为 *)
     * @param messageClass
     *            消息实体类(必须为 Object)
     * @param listener
     *            消息监听回调
     * @param <T>
     *            消息实体类型
     */
    private <T> void subscribe(@NonNull String gid, @NonNull String nameServerAddress, @NonNull String topic,
        @Nullable String tag, @NonNull Class<T> messageClass, @NonNull MqListener<T> listener) {
        if (StringUtils.isEmpty(tag)) {
            tag = "*";
        }
        Properties properties = new Properties();
        // 您在控制台创建的 Group ID
        properties.put(PropertyKeyConst.GROUP_ID, gid);
        // 鉴权用 AccessKey，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.AccessKey, rocketMqProperties.getAccessKey());
        // 鉴权用 SecretKey，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.SecretKey, rocketMqProperties.getSecretKey());
        // 设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看
        properties.put(PropertyKeyConst.NAMESRV_ADDR, nameServerAddress);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(topic, tag, new CpxMessageListener<>(listener, messageClass));
        consumer.start();

        log.info("RocketMq Consumer Started. topic: {}, gid: {}, nameServerAddress: {}",
            String.format("%s:%s", topic, tag), gid, nameServerAddress);
    }

    /**
     * 获取 {@link MqListener} 子类的的 消息实体类
     *
     * @param redisMQListener
     *            {@link MqListener} 子类实例
     * @return 消息实体类
     */
    private Class getMessageClass(MqListener redisMQListener) {
        Class listenerClass = MqListener.class;
        Class<?> beanClass = redisMQListener.getClass();

        ParameterizedType parameterizedType = getParameterizedType(listenerClass, beanClass.getGenericInterfaces());
        Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
        Class messageClass;
        try {
            messageClass = (Class)actualTypeArguments[0];
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new IllegalArgumentException(
                listenerClass.getName() + "实现类泛型类型参数列表必须只有一个Class参数.eg:MqListener<Object>", e);
        }
        return messageClass;
    }

    /**
     * 从实体类的继承的所有接口获取预期继承的接口类型的参数化类型
     *
     * @param targetClass
     *            预期继承的接口类型
     * @param genericInterfaces
     *            实体类的继承的所有接口
     * @return 预期继承的接口类型的参数化类型
     */
    private <T> ParameterizedType getParameterizedType(Class<T> targetClass, Type[] genericInterfaces) {
        List<Type> interfaceList = Arrays.stream(genericInterfaces).filter(type -> {
            if (ParameterizedType.class.isAssignableFrom(type.getClass())) {
                ParameterizedType parameterizedType = (ParameterizedType)type;
                return parameterizedType.getRawType() == targetClass;
            }
            return false;
        }).collect(Collectors.toList());
        if (interfaceList.size() == 1) {
            return (ParameterizedType)interfaceList.get(0);
        }
        return null;
    }

}
